package com.oracle.usercf;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class Job4 {
	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String split[] = line.split("\t");
			
				if(line.substring(line.indexOf("\t")+1, line.indexOf("_")).matches("[0-9]")) {
					for(int i = 1; i < split.length; i++) {
						context.write(new Text(split[0]), new Text(split[i]));
					}
				}else {
					for(int i = 1; i < split.length; i++) {
						context.write(new Text(split[i].substring(0, split[i].indexOf("_"))), new Text(split[0]+split[i].substring(split[i].indexOf("_"))));
					}
				}
			
			
			
		}
	}
	
	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			List<String> list_left = new ArrayList<>();
			List<String> list_right = new ArrayList<>();
			
			for(Text i: values) {
				if(i.toString().substring(0, i.toString().indexOf("_")).matches("[0-9]")) {
					list_right.add(i.toString());
				}else {
					list_left.add(i.toString());
				}
			}
			
			String row;
			double row_val;
			String col;
			double col_val;
			double val;
			for(String i: list_left) {
				row = i.substring(0,i.indexOf("_"));
				row_val = Double.parseDouble(i.substring(i.indexOf("_")+1));
				for(String j: list_right) {
					col = j.substring(0,i.indexOf("_"));
					col_val = Double.parseDouble(j.substring(j.indexOf("_")+1));
					val = row_val*col_val;
					
					context.write(new Text(row+","+col), new Text(String.format("%.2f", val)));
				}
			}
		
			
		
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		
		Job job = Job.getInstance(conf);
		job.setJarByClass(Job1.class);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		FileInputFormat.setInputPaths(job, new Path("/usercf/output1"));
		FileInputFormat.addInputPath(job, new Path("/usercf/output3"));
		FileOutputFormat.setOutputPath(job, new Path("/usercf/output4"));
		
		job.waitForCompletion(true);
	}
}
